package drds.global_transaction.api.rpc.rpc_client;

import drds.common.Log;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerResponseMessage;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool.KeyedPoolableObjectFactory;

import java.net.InetSocketAddress;

@Slf4j
public class CreateChannelFactory implements KeyedPoolableObjectFactory<Key, Channel>
{
	private final RpcClient rpcClient;

	public CreateChannelFactory(RpcClient rpcClient)
	{
		this.rpcClient = rpcClient;
		this.rpcClient.setChannelHandlers(new ChannelHandler[] { rpcClient });
		this.rpcClient.start();
	}

	@Override
	public Channel makeObject(Key key) throws Exception
	{
		InetSocketAddress inetSocketAddress = Nets.toInetSocketAddress(key.getAddress());
		Channel channel = rpcClient.connect(inetSocketAddress);
		Object response = null;
		Channel needReturnChannel = null;
		try
		{
			response = rpcClient.asynchronousSendRequestWithDefaultTimeoutGetResponse(null, channel, key.getMessage());
			if (!isRegisterSuccess(key.getRole(), response))
			{
				rpcClient.onRegisterMessageFail(key.getAddress(), channel, key.getMessage(), response);
				log.warn("服务器端版本:" + getVersion(response, key.getRole()));
			} else
			{
				needReturnChannel = channel;
				rpcClient.onRegisterMessageSuccess(key.getAddress(), channel, key.getMessage(), response);
			}
		} catch (Exception e)
		{
			if (channel != null)
			{
				try
				{
					channel.disconnect();
				} catch (Exception e1)
				{
					e1.printStackTrace();
				}
				channel.close();
			}
			Log.error(log, Thread.currentThread(), "", e);
			throw new FrameworkException("register error,role:" + key.getRole().name() + ",err:" + e.getMessage());
		}
		return needReturnChannel;
	}

	private boolean isRegisterSuccess(Role role, Object response)
	{
		if (null == response)
		{
			return false;
		}
		if (role.equals(Role.transaction_manager_client))
		{
			if (!(response instanceof RegisterTransactionManagerResponseMessage))
			{
				return false;
			}
			if (((RegisterTransactionManagerResponseMessage) response).isSuccess())
			{
				return true;
			}
		} else if (role.equals(Role.resource_manager_client))
		{
			if (!(response instanceof RegisterResourceManagerResponseMessage))
			{
				return false;
			}
			if (((RegisterResourceManagerResponseMessage) response).isSuccess())
			{
				return true;
			}
		}
		return false;
	}

	private String getVersion(Object response, Role role)
	{
		if (role.equals(Role.transaction_manager_client))
		{
			return ((RegisterTransactionManagerResponseMessage) response).getVersion();
		} else
		{
			return ((RegisterResourceManagerResponseMessage) response).getVersion();
		}
	}

	@Override
	public void activateObject(Key key, Channel obj) throws Exception
	{
	}

	@Override
	public void passivateObject(Key key, Channel obj) throws Exception
	{

	}

	@Override
	public boolean validateObject(Key key, Channel channel)
	{
		if (null != channel && channel.isActive())
		{
			return true;
		} else
		{
			return false;
		}
	}

	@Override
	public void destroyObject(Key key, Channel channel) throws Exception
	{

		if (null != channel)
		{
			channel.disconnect();
			channel.close();
		}
	}

}
